use bytes::Bytes;
use common::base::status::Status;
use common::service::sys_service_api::BoolResponse;
use common::service::sys_service_api::db::DBExecuteResult;

use crate::{application::APP, db::mysql::MySQLConnect, service::service::Service};

use super::{
    get_local_service, pack_and_return_error, pack_and_return_message,
    pack_and_return_status_and_bytes, pack_and_return_success,
};

pub extern "C" fn get_connection(service_id: i64, request_id: i64) {
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();

        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }
        let table_prefix = format!(
            "{}_{}",
            service_key.dxc_name,
            service_key.dxc_version.replace(".", "_")
        );

        let mysql_pool = mysql_pool.as_ref().unwrap();

        mysql_pool.new_connection(request_id, table_prefix);

        pack_and_return_success(request_id, service_key)
    });
}

pub extern "C" fn putback_connection(service_id: i64, request_id: i64) {
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();

        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }

        let mysql_pool = mysql_pool.as_ref().unwrap();

        mysql_pool.putback_connection(request_id);

        pack_and_return_success(request_id, service_key)
    });
}

pub extern "C" fn begin_tx(service_id: i64, request_id: i64) {
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();

        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }
        let mysql_pool = mysql_pool.as_ref().unwrap();
        let connection = mysql_pool.get_connection(request_id);
        if connection.is_none() {
            pack_and_return_error(request_id, service_key, Status::error("链接不存在".into()));
            return;
        }

        connection.unwrap().lock().await.begin_tx().await;

        pack_and_return_success(request_id, service_key)
    });
}

pub extern "C" fn rollback(service_id: i64, request_id: i64) {
    rollback_or_commit(service_id, request_id, true);
}

pub extern "C" fn commit(service_id: i64, request_id: i64) {
    rollback_or_commit(service_id, request_id, false);
}

fn rollback_or_commit(service_id: i64, request_id: i64, is_rollback: bool) {
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();

        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }

        let mysql_pool = mysql_pool.as_ref().unwrap();

        let connection: Option<std::sync::Arc<tokio::sync::Mutex<MySQLConnect<'_>>>> =
            mysql_pool.get_connection(request_id);

        if connection.is_none() {
            pack_and_return_error(request_id, service_key, Status::error("事务不存在".into()));
            return;
        }

        let result = if is_rollback {
            connection.unwrap().lock().await.rollback().await
        } else {
            connection.unwrap().lock().await.commit().await
        };

        if result.is_err() {
            pack_and_return_error(request_id, service_key, result.err().unwrap());
        } else {
            pack_and_return_success(request_id, service_key);
        }
    });
}

/**
 *
 */
pub extern "C" fn is_table_exist(service_id: i64, request_id: i64, sql: *const u8, len: u32) {
    let table_name = unsafe {
        let slice = std::slice::from_raw_parts(sql, len as usize);
        std::str::from_utf8_unchecked(slice).to_string()
    };
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();
        //
        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }
        //
        let mysql_pool = mysql_pool.as_ref().unwrap();

        let connection = mysql_pool.get_connection(request_id);

        if connection.is_none() {
            pack_and_return_error(request_id, service_key, Status::error("连接不存在".into()));
            return;
        }
        //
        let result = connection
            .unwrap()
            .lock()
            .await
            .is_table_exist(&table_name)
            .await;

        if result.is_err() {
            pack_and_return_error(request_id, service_key, result.err().unwrap());
        } else {
            let is_exist = result.unwrap();

            let mut result = BoolResponse::default();

            result.is_exist = is_exist;

            pack_and_return_message(request_id, service_key, result);
        }
    });
    //
}

/**
 *
 */
pub extern "C" fn query(service_id: i64, request_id: i64, sql: *const u8, len: u32) {
    let str_sql = unsafe {
        let slice = std::slice::from_raw_parts(sql, len as usize);
        std::str::from_utf8_unchecked(slice).to_string()
    };
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();
        //
        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }
        //
        let mysql_pool = mysql_pool.as_ref().unwrap();

        let connection = mysql_pool.get_connection(request_id);

        if connection.is_none() {
            pack_and_return_error(request_id, service_key, Status::error("连接不存在".into()));
            return;
        }
        //
        let result = connection.unwrap().lock().await.query(&str_sql).await;
        if result.is_err() {
            pack_and_return_error(request_id, service_key, result.err().unwrap());
        } else {
            let buffer = result.unwrap();
            let bytes = Bytes::from(buffer);
            pack_and_return_status_and_bytes(request_id, service_key, Status::default(), bytes);
        }
    });
}

pub extern "C" fn execute(service_id: i64, request_id: i64, sql: *const u8, len: u32) {
    let str_sql = unsafe {
        let slice = std::slice::from_raw_parts(sql, len as usize);
        std::str::from_utf8_unchecked(slice).to_string()
    };
    tokio::spawn(async move {
        let local_service = get_local_service(service_id, request_id, false).await;
        if local_service.is_none() {
            return;
        }
        let local_service = local_service.unwrap();
        //
        let service_key = local_service.get_service_key();
        let mysql_pool = unsafe { &APP.assume_init_ref().mysql_pool };
        if mysql_pool.is_none() {
            pack_and_return_error(
                request_id,
                service_key,
                Status::error("节点不支持 MySQL 操作".into()),
            );
            return;
        }
        //
        let mysql_pool = mysql_pool.as_ref().unwrap();
        let connection = mysql_pool.get_connection(request_id);
        if connection.is_none() {
            pack_and_return_error(request_id, service_key, Status::error("连接不存在".into()));
            return;
        }
        //
        let result = connection.unwrap().lock().await.execute(&str_sql).await;
        if result.is_err() {
            pack_and_return_error(request_id, service_key, result.err().unwrap());
        } else {
            let (last_insert_id, affected_rows) = result.unwrap();
            let mut result = DBExecuteResult::default();
            result.last_insert_id = last_insert_id;
            result.affected_rows = affected_rows;
            pack_and_return_message(request_id, service_key, result);
        }
    });
}
